package udf.matching;

import java.io.IOException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;

import model.Collocation;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import udf.string.CleanTag;

public class generateQueryTags extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();
	CleanTag cleaner = null;

	private boolean initialized = false;

	private Hashtable<String, Collocation> hyphenized = new Hashtable<String, Collocation>();
	//private Hashtable<String, String> simple = new Hashtable<String, String>();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	/**
	 * Tuple fields
	 * 
	 * 0 set of tags of the bookmark 1 set of tags hpyhenized (dictionary)
	 * 
	 * 
	 * @param tuple
	 * @throws IOException
	 */

	/**
	 * Position 0 : a_b Position 1 : freq
	 * 
	 * 
	 */

	
	
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(4);

		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();

			String tag = (String) tup.get(0);
			Integer f = (Integer) tup.get(1);

			Float pmi = (Float) tup.get(2);
			Float ratio = (Float) tup.get(3);

			Collocation col = new Collocation();
			col.freq_h = f;
			//col.pmi = pmi;
			//col.ratio = ratio;

		//	 System.out.println("Adding: " + tag + "\t" +f);
			hyphenized.put(tag, col);
			// System.out.dfprintln("Init dictionary:" + tag + "\t"+ freq);
			// add simple
		
	

		}

		initialized = true;
	}

	
	
	/**
	 * Improvement: ADD PMI RATIO THRESHOLD TO DECIDE WHEN TO ADD BIGRAM
	 * OR WHEN TO ADD BIGRAM AND ADD SINGLE TOKENS AS WELL TO THE HASH
	 * 
	 * 
	 * 
	 * @param query
	 * @param title
	 * @param description
	 * @param threshold
	 * @return
	 */
	
	public Hashtable<String, Integer> getTags(String query, String title,
			String description, int threshold) {
		Hashtable<String, Integer> suggestions = new Hashtable<String, Integer>();

		
		//System.out.println("Tags of  query:" +query);
		LinkedList<String> temporal = splitText(query);
		//for(int i =0; i< temporal.size(); i++){
	
		//	System.out.println(temporal.get(i));
		//}
		updateHash(suggestions, temporal, threshold);

		//System.out.println("Hash frequency:");
		//printHash(suggestions);
		
		
		temporal = splitText(title);

		//System.out.println("Tags of  title:" +title);
		
		///for(int i =0; i< temporal.size(); i++){
	
		//	System.out.println(temporal.get(i));
		//}
		//System.out.println("Hash frequency:");
		
		updateHash(suggestions, temporal, threshold);
				
		//printHash(suggestions);
		
		

		temporal = splitText(description);

		updateHash(suggestions, temporal, threshold);

		return suggestions;
	}
	
	
	public void printHash(Hashtable<String, Integer> suggestions){
		
		Enumeration<String> keys = suggestions.keys();
		
		while(keys.hasMoreElements()){
			String key = keys.nextElement();
			System.out.println(key + "\t"+ suggestions.get(key));
			
		}
	}

	/**
	 * try to match bigrams in text
	 * 
	 * if bigram is match it is added directly
	 * 
	 * otherwise each term is added independently
	 * 
	 * @param suggestions
	 * 
	 * @param temporal
	 */

	private void updateHash(Hashtable<String, Integer> suggestions,
			LinkedList<String> temporal, int THRESHOLD) {
		// TODO Auto-generated method stub
		String a = null;
		String b = null;

		for (int i = 0; i < temporal.size() - 2; i++) {
			a = temporal.get(i);

			b = temporal.get(i + 1);
			int freq = 1;

			// try bigrams
			String a_b = a + "_" + b;
			a_b= a_b.toLowerCase();

			if (hyphenized.containsKey(a_b)
					&& hyphenized.get(a_b).freq_h >= THRESHOLD) {

				if (suggestions.containsKey(a_b)) {
					freq = suggestions.get(a_b) + 1;

				}

				suggestions.put(a_b, freq);

			} else {

				if (suggestions.containsKey(a)) {
					freq = suggestions.get(a) + 1;

				}
				suggestions.put(a, freq);

			}

		}
		int freq = 1;
		if (temporal.size() - 2 >= 0) {
			
			a = temporal.get(temporal.size() - 2);
			b = temporal.get(temporal.size() - 1);

			// last bigram is a special case
			String a_b = a + "_" + b;
			
			if (hyphenized.containsKey(a_b)) {

				if (suggestions.containsKey(a_b)) {
					freq = suggestions.get(a_b) + 1;

				}
				suggestions.put(a_b, freq);

			} else {

				if (suggestions.containsKey(a)) {
					freq = suggestions.get(a) + 1;

				}
				suggestions.put(a, freq);
				freq=1;
				if (suggestions.containsKey(b)) {
					freq = suggestions.get(b) + 1;

				}
				suggestions.put(b, freq);

			}

		} else {
			freq=1;
			if(temporal.size() - 1 >= 0){
				b = temporal.get(temporal.size() - 1);
				if (suggestions.containsKey(b)) {
					freq = suggestions.get(b) + 1;

				}
				
				suggestions.put(b, freq);
				
			}

		}

	}

	public LinkedList<String> splitText(String text) {

		LinkedList<String> suggestions = new LinkedList<String>();

		if (text == null) {

			return suggestions;
		}

		text = text.replaceAll("\\s+", " ").toLowerCase();
		String words[] = text.split(" ");
		for (int i = 0; i < words.length; i++) {
			
			String word = CleanTag.normalizeTag(words[i]);

			if (!cleaner.isTagTrash(word)) {
				suggestions.add(word);

			}

		}

		return suggestions;

	}

	public DataBag exec(Tuple input) throws IOException {

		
		if (!initialized)
			init(input);
		
		
		if (input.get(0) == null) {
			return null;
		}

		else {

			String path = (String) input.get(3);
			String query = (String) input.get(0);
			String title = (String) input.get(1);
			String snippet = (String) input.get(2);

			int threshold = input.getType(5);
			// System.out.println(query+"\t" + title+ "\t"+ snippet);

			if (cleaner == null) {
				cleaner = new CleanTag(path);

			}

			DataBag output = mBagFactory.newDefaultBag();

			Hashtable<String, Integer> list = getTags(query, title, snippet,
					threshold);

			if (list == null) {
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append("-error-");
				tuple.append(-1);
				output.add(tuple);

				// System.out.println(query + "\t" + "-error-" + "\t" + (-1));

			} else if (list.size() == 0) {
				TupleFactory mTupleFactory = TupleFactory.getInstance();
				Tuple tuple = mTupleFactory.newTuple();
				tuple.append("-no_tags-");
				tuple.append(-2);
				output.add(tuple);

			} else {
				Enumeration<String> iter = list.keys();

				while (iter.hasMoreElements()) {
					String tag = iter.nextElement();
					TupleFactory mTupleFactory = TupleFactory.getInstance();
					Tuple tuple = mTupleFactory.newTuple();
					tuple.append(tag);
					tuple.append(list.get(tag));
					output.add(tuple);

				}

			}

			// StringTokenizer tok = new StringTokenizer((String)o,
			// " \",()*", false);
			// while (tok.hasMoreTokens())
			// output.add(mTupleFactory.newTuple(tok.nextToken()));
			// return output;

			return output;

		}

	}

	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();

			bagSchema.add(new Schema.FieldSchema("suggestion",
					DataType.CHARARRAY));

			bagSchema.add(new Schema.FieldSchema("freq", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}
}
